package com.ruckuswireless.pentaho.protobuf.encode;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.RowMeta;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMeta;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

import com.google.protobuf.Descriptors.FieldDescriptor;

/**
 * Main processing unit for the Protocol Buffers Encode step
 * 
 * @author Michael Spector
 */
public class ProtobufEncodeStep extends BaseStep implements StepInterface {

	private Map<String, FieldDescriptor> name_fdMap = new HashMap<String, FieldDescriptor>();

	public ProtobufEncodeStep(StepMeta stepMeta,
			StepDataInterface stepDataInterface, int copyNr,
			TransMeta transMeta, Trans trans) {
		super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
	}

	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
		super.init(smi, sdi);

		ProtobufEncodeMeta meta = (ProtobufEncodeMeta) smi;
		ProtobufEncodeData data = (ProtobufEncodeData) sdi;
		try {
			List<FieldDescriptor> fields = com.run.pbf.Object.ObjectRelation
					.getDescriptor().getFields();
			for (FieldDescriptor fd : fields) {
				logBasic("init fd.getName(" + fd.getName() + ")");
				if (fd.getName().equals("src_ip")) {
					logBasic("init src_ip");
					name_fdMap.put(fd.getName(), fd);
				}
				if (fd.getName().equals("data_source")) {
					logBasic("init data_source");
					name_fdMap.put(fd.getName(), fd);
				}
			}
		} catch (Exception e) {
			logError(Messages.getString("ProtobufEncodeStep.Init.Error",
					getStepname()), e);
			return false;
		}
		return true;
	}

	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi)
			throws KettleException {
		try {
			Object[] r = getRow();
			if (r == null) {
				setOutputDone();
				return false;
			}
			ProtobufEncodeMeta meta = (ProtobufEncodeMeta) smi;
			ProtobufEncodeData data = (ProtobufEncodeData) sdi;
			RowMetaInterface inputRowMeta = getInputRowMeta();
			if (first) {
				data.outputRowMeta = getInputRowMeta().clone();

				RowMetaInterface newFields = new RowMeta();

				ValueMetaInterface newFieldMeta = new ValueMeta("pbf",
						ValueMeta.TYPE_BINARY);
				newFields.addValueMeta(newFieldMeta);

				data.outputRowMeta.addRowMeta(newFields);
			}

			int indexOfValue_src_ip = inputRowMeta.indexOfValue("src_ip");
			int indexOfValue_ds = inputRowMeta.indexOfValue("data_source");
			Long src_ip = (Long) r[indexOfValue_src_ip];
			Long data_source = (Long) r[indexOfValue_ds];
			logBasic("==== src_ip[" + src_ip + "]");
			logBasic("==== ds[" + data_source + "]");
			com.run.pbf.Object.ObjectRelation.Builder orb = com.run.pbf.Object.ObjectRelation
					.newBuilder();
			orb.setField(name_fdMap.get("src_ip"), src_ip);
			orb.setField(name_fdMap.get("data_source"), data_source.intValue());
			r = RowDataUtil.addValueData(r, inputRowMeta.size(), orb.build()
					.toByteArray());
			putRow(data.outputRowMeta, r);

		} catch (Exception e) {
			logError(e.getMessage(), e);
			setErrors(1);
			stopAll();
			setOutputDone();
			return false;
			// putError(inputRowMeta, r, 1, e.toString(), null, getStepname());
		}
		return true;
	}

	public void stopRunning(StepMetaInterface smi, StepDataInterface sdi)
			throws KettleException {

		ProtobufEncodeData data = (ProtobufEncodeData) sdi;
		data.canceled = true;

		super.stopRunning(smi, sdi);
	}
}
